[FLINK]Support processing time tumble window for nexmark q12#13
Open
KevinyhZou wants to merge 19 commits intobigo-sg:gluten-0530from
Open
[FLINK]Support processing time tumble window for nexmark q12#13KevinyhZou wants to merge 19 commits intobigo-sg:gluten-0530from
KevinyhZou wants to merge 19 commits intobigo-sg:gluten-0530from
Conversation
KevinyhZou
changed the title
Merged
zhanglistar
reviewed
Dec 2, 2025
| // Look for the end of the last group. | ||
| vector_size_t index = 0; | ||
| if (prevInput_) { | ||
| if (prevInput_ && numGroups_ > 0) { |
zhanglistar
reviewed
Dec 2, 2025
| RowVectorPtr output; | ||
|
|
||
| if (numGroups_ > minOutputBatchSize_) { | ||
| if (numGroups_ >= minOutputBatchSize_) { |
Author
There was a problem hiding this comment.
对，velox 里面minOutputBatchSize 最小只能取1， numGroups > minOutputBatchSize 意味着当只有一条数据的时候，是不输出的，这不符合流计算的场景。
zhanglistar
reviewed
Dec 2, 2025
| } | ||
| } | ||
|
|
||
| long TimeWindowUtil::getCurrentProcessingTime() { |
zhanglistar
reviewed
Dec 2, 2025
|
|
||
| static long cleanupTime(long maxTimestamp, long allowedLateness_, bool isEventTime); | ||
|
|
||
| static long getCurrentProcessingTime(); |
Member
There was a problem hiding this comment.
static int64_t getCurrentProcessingTime(); since sizeof(long) may 4 bytes on 32 bit machine.
zhanglistar
reviewed
Dec 2, 2025
| now.time_since_epoch()).count(); | ||
| return {{timestamp_ms, input}}; | ||
| long timestamp_ms = TimeWindowUtil::getCurrentProcessingTime(); | ||
| if (windowType_ == 1) { // tumble window |
zhanglistar
reviewed
Dec 2, 2025
| ProcessingTimeCallback callback_; | ||
| }; | ||
|
|
||
| class ProcessingTimeSerivice { |
zhanglistar
reviewed
Dec 2, 2025
| return TimeWindowUtil::getCurrentProcessingTime(); | ||
| } | ||
| virtual std::optional<std::string> registerTimer(long timestamp, ProcessingTimerTask target) { | ||
| std::optional<std::string> task; |
Member
There was a problem hiding this comment.
用纯虚函数，virtual std::optional<std::string> registerTimer(long timestamp, ProcessingTimerTask target) = 0;
zhanglistar
reviewed
Dec 2, 2025
| virtual void cancel(const std::string& task) {} | ||
| virtual void close() {} | ||
|
|
||
| void finish(const std::string& task) { |
zhanglistar
reviewed
Dec 2, 2025
| return "proc_time_task_" + std::to_string(timestamp); | ||
| } | ||
| protected: | ||
| std::vector<std::string> registry; |
Member
There was a problem hiding this comment.
用std::set是不是更好？ vector不适合频繁查找和删除。
zhanglistar
reviewed
Dec 2, 2025
|
|
||
| namespace facebook::velox::stateful { | ||
|
|
||
| using ProcessingTimeCallback = std::function<void(long)>; |
Member
There was a problem hiding this comment.
long统一都改成int64_t吧 避免32位系统的溢出
zhanglistar
reviewed
Dec 2, 2025
| } | ||
|
|
||
| std::string generateTimerTaskName(long timestamp) { | ||
| return "proc_time_task_" + std::to_string(timestamp); |
Member
There was a problem hiding this comment.
如果在统一ms注册 这个可能冲突？加个counter
static std::atomic<uint64_t> counter{0};
return "proc_time_task_" + std::to_string(timestamp) + "_" + std::to_string(counter++);
zhanglistar
reviewed
Dec 2, 2025
| void close() override { | ||
| if (executor_) { | ||
| executor_->shutdown(); | ||
| } |
zhanglistar
reviewed
Dec 2, 2025
|
|
||
| /// This class is relevent to flink KeySelector. | ||
| /// It can partition the RowVector according to the key fields. | ||
| class KeySelector { |
Member
There was a problem hiding this comment.
KeySelector 应改是KeyPartitioner？
zhanglistar
reviewed
Dec 2, 2025
| @@ -31,7 +31,7 @@ | |||
| memory::MemoryPool* pool, | |||
| int numPartitions = INT_MAX); | |||
zhanglistar
reviewed
Dec 2, 2025
| } | ||
|
|
||
| std::map<uint32_t, RowVectorPtr> KeySelector::partition(const RowVectorPtr& input) { | ||
| std::map<uint64_t, RowVectorPtr> KeySelector::partition(const RowVectorPtr& input) { |
Member
There was a problem hiding this comment.
这个接口有点重，不应该用map 不过跟你这次改动没关系了 后面再改吧
zhanglistar
reviewed
Dec 2, 2025
| } | ||
| } | ||
|
|
||
| void KeySelector::allocateIndexBuffers( |
Member
There was a problem hiding this comment.
这个接口实现重，这样更好
struct IndexBuffers {
std::vector<BufferPtr> buffers;
std::vector<vector_size_t*> rawPtrs;
};
IndexBuffers allocateIndexBuffers(const std::vector<vector_size_t>& counts);
zhanglistar
reviewed
Dec 2, 2025
| #include "velox/experimental/stateful/Triggerable.h" | ||
| #include "velox/experimental/stateful/window/SliceAssigner.h" | ||
| #include "velox/experimental/stateful/window/WindowBuffer.h" | ||
| #include <memory> |
zhanglistar
reviewed
Dec 2, 2025
| StatefulOperator::close(); | ||
| localAggerator_->close(); | ||
| if (localAggerator_) { | ||
| localAggerator_->close(); |
zhanglistar
reviewed
Dec 2, 2025
| #include "velox/experimental/stateful/window/TimeWindowUtil.h" | ||
|
|
||
| #include <list> | ||
| #include <memory> |
zhanglistar
reviewed
Dec 2, 2025
| StatefulOperator::initialize(); | ||
| localAggerator_->initialize(); | ||
| if (localAggerator_) { | ||
| localAggerator_->initialize(); |
zhanglistar
reviewed
Dec 2, 2025
| const int windowStartIndex, | ||
| const int windowEndIndex) | ||
| : StatefulOperator(std::move(globalAggerator), std::move(targets)), | ||
| localAggerator_(std::move(localAggerator)), |
zhanglistar
reviewed
Dec 2, 2025
| @@ -149,7 +272,9 @@ long WindowAggregator::sliceStateMergeTarget(long sliceToMerge) { | |||
| void WindowAggregator::close() { | |||
| processWatermarkInternal(INT_MAX); | |||
Member
There was a problem hiding this comment.
processWatermarkInternal(std::numeric_limits<int64_t>::max());
zhanglistar
reviewed
Dec 2, 2025
| const RowVectorPtr& output, | ||
| const std::string& fieldName, | ||
| const TypePtr& fieldType, | ||
| const long fieldValue, |
zhanglistar
reviewed
Dec 2, 2025
| RowVectorPtr addWindowTimestampToOutput( | ||
| const RowVectorPtr& output, | ||
| const std::string& fieldName, | ||
| const TypePtr& fieldType, |
zhanglistar
reviewed
Dec 2, 2025
| std::list<RowVectorPtr> allDatas; | ||
| for (const auto& data: datas) { | ||
| allDatas.push_back(data); | ||
| } |
Member
There was a problem hiding this comment.
std::vector<RowVectorPtr> allDatas(datas.begin(), datas.end());
zhanglistar
reviewed
Dec 2, 2025
|
|
||
| template<typename K> | ||
| void WindowAggregator::fireWindow(K key, long timerTimestamp, long windowEnd) { | ||
| RowVectorPtr output = windowState_->value(key, windowEnd); |
Member
There was a problem hiding this comment.
if (!output) {
LOG(INFO) << "No output found for key: " << key << ", window end: " << windowEnd;
return;
}
zhanglistar
reviewed
Dec 2, 2025
| windowState_->remove(key, windowEnd); | ||
| } | ||
|
|
||
| void WindowAggregator::onEventTime(std::shared_ptr<TimerHeapInternalTimer<uint32_t, long>> timer) { |
zhanglistar
reviewed
Dec 2, 2025
| onTimer(timer); | ||
| } | ||
|
|
||
| void WindowAggregator::onProcessingTime(std::shared_ptr<TimerHeapInternalTimer<uint32_t, long>> timer) { |
zhanglistar
reviewed
Dec 2, 2025
|
|
||
| template<typename K> | ||
| void WindowAggregator::clearWindow(K key, long timerTimestamp, long windowEnd) { | ||
| windowState_->remove(key, windowEnd); |
zhanglistar
reviewed
Dec 2, 2025
| size_ = 0; | ||
| } | ||
|
|
||
| bool empty() override { |
Member
There was a problem hiding this comment.
bool empty() const override
lgbo-ustc
reviewed
Jan 6, 2026
|
|
||
| namespace facebook::velox::stateful { | ||
|
|
||
| static int roundUpToPowerOfTwo(int32_t x) { |
There was a problem hiding this comment.
参照ck的实现
template <typename T>
requires std::is_integral_v<T> && (sizeof(T) == sizeof(UInt64))
inline T roundDownToPowerOfTwo(T x)
{
return x <= 0 ? 0 : (T(1) << (63 - __builtin_clzll(x)));
}
lgbo-ustc
reviewed
Jan 9, 2026
| int numPartitions = 1024); | ||
|
|
||
| std::map<uint32_t, RowVectorPtr> partition(const RowVectorPtr& input); | ||
| std::map<int64_t, RowVectorPtr> partition(const RowVectorPtr& input); |
There was a problem hiding this comment.
key使用一个类型别名吧，估计后面需要换类型，64位hash值还是有可能会出现冲突。map换 unordered_map
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
StreamingAggreagate